package com.demo.controllers;

import com.demo.customresources.Rediscluster;
import io.fabric8.kubernetes.api.model.OwnerReference;
import io.fabric8.kubernetes.api.model.Pod;
import io.fabric8.kubernetes.api.model.Service;
import io.fabric8.kubernetes.api.model.apps.Deployment;
import io.fabric8.kubernetes.client.Callback;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.client.KubernetesClientException;
import io.fabric8.kubernetes.client.dsl.ExecWatch;
import io.fabric8.kubernetes.client.dsl.RollableScalableResource;
import io.fabric8.kubernetes.client.dsl.ServiceResource;
import io.fabric8.kubernetes.client.utils.InputStreamPumper;
import io.fabric8.kubernetes.client.utils.Serialization;
import io.javaoperatorsdk.operator.api.*;
import io.javaoperatorsdk.operator.processing.event.EventSourceManager;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Controller
public class RedisClusterController implements ResourceController<Rediscluster> {

    private static final Logger log = LoggerFactory.getLogger(RedisClusterController.class);

    private final KubernetesClient kubernetesClient;

    public RedisClusterController(KubernetesClient kubernetesClient) {
        this.kubernetesClient = kubernetesClient;
    }
    //缓存pod ip
    private static Map<String,List<String>> cacheIPs=new ConcurrentHashMap<>();

    //删除后清理自动创建的deployment和service资源
    @Override
    public DeleteControl deleteResource(Rediscluster rediscluster, Context<Rediscluster> context) {
        log.info("Execution deleteResource for: {}", rediscluster.getMetadata().getName());

        //清除pod ip缓存
        cacheIPs.remove(rediscluster.getMetadata().getName());


        //删除service
        try {
            this.deleteService(rediscluster);
        } catch (Exception e) {
            log.error("", e);
        }

        //删除deployment
        try {
            this.deleteDeployment(rediscluster);
        } catch (Exception e) {
            log.error("", e);
        }

        return DeleteControl.DEFAULT_DELETE;
    }

    @Override
    public void init(EventSourceManager eventSourceManager) {
        log.info("Execution init for RedisClusterController");
    }

    @Override
    public UpdateControl<Rediscluster> createOrUpdateResource(Rediscluster rediscluster, Context<Rediscluster> context) {
        log.info("Execution createOrUpdateResource for: {}", rediscluster.getMetadata().getName());


        //创建或更新k8s service
        try {
            this.createOrUpdateService(rediscluster);
        } catch (Exception e) {
            log.error("", e);
        }
        //创建或更新k8s deployment
        try {
            this.createOrUpdateDeployment(rediscluster);
        } catch (Exception e) {
            log.error("", e);
        }

        return UpdateControl.updateCustomResource(rediscluster);
    }


    //创建或更新的k8s deployment
    private void createOrUpdateDeployment(Rediscluster rediscluster) throws Exception {
        String ns = rediscluster.getMetadata().getNamespace();
        Deployment existingDeployment =
                kubernetesClient
                        .apps()
                        .deployments()
                        .inNamespace(ns)
                        .withName(rediscluster.getMetadata().getName())
                        .get();
        if (existingDeployment == null) {
            Deployment deployment = loadYaml(Deployment.class, "/redis-cluster-deployment-template.yml");
            deployment.getMetadata().setName(rediscluster.getMetadata().getName());
            deployment.getMetadata().setNamespace(ns);
            deployment.getMetadata().getLabels().put("app.kubernetes.io/part-of", rediscluster.getMetadata().getName());
            deployment.getMetadata().getLabels().put("app.kubernetes.io/managed-by", "redisclusteroperator");
            // set tomcat version
            deployment
                    .getSpec()
                    .getTemplate()
                    .getSpec()
                    .getContainers()
                    .get(0)
                    .setImage(rediscluster.getSpec().getImage())
            ;//镜像
            deployment.getSpec().setReplicas(rediscluster.getSpec().getPodNum());//redis节点数

            // make sure label selector matches label (which has to be matched by service selector too)
            deployment
                    .getSpec()
                    .getTemplate()
                    .getMetadata()
                    .getLabels()
                    .put("app", rediscluster.getMetadata().getName());
            deployment
                    .getSpec()
                    .getSelector()
                    .getMatchLabels()
                    .put("app", rediscluster.getMetadata().getName());

            //设置密码

            List<String> passwordCommand = List.of("/bin/sh", "-ce", "redis-cli config set requirepass " + rediscluster.getSpec().getPassword() + " && redis-cli config set masterauth  " + rediscluster.getSpec().getPassword());
            log.info(StringUtils.join(passwordCommand, " "));
            deployment
                    .getSpec()
                    .getTemplate()
                    .getSpec()
                    .getContainers()
                    .get(0).getLifecycle().getPostStart().getExec().setCommand(passwordCommand);


            OwnerReference ownerReference = deployment.getMetadata().getOwnerReferences().get(0);
            ownerReference.setName(rediscluster.getMetadata().getName());
            ownerReference.setUid(rediscluster.getMetadata().getUid());

            log.info("Creating or updating Deployment {} in {}", deployment.getMetadata().getName(), ns);
            //创建deployment
            kubernetesClient.apps().deployments().inNamespace(ns).create(deployment);

            //根据pod列表创建集群
            this.createRedisCluster(ns, rediscluster);

        } else {
            existingDeployment
                    .getSpec()
                    .getTemplate()
                    .getSpec()
                    .getContainers()
                    .get(0)
                    .setImage(rediscluster.getSpec().getImage());//镜像
            existingDeployment.getSpec().setReplicas(rediscluster.getSpec().getPodNum());//redis节点数

            log.info("updating Deployment {} in {}", existingDeployment.getMetadata().getName(), ns);
            kubernetesClient.apps().deployments().inNamespace(ns).createOrReplace(existingDeployment);

            Thread.sleep(10000);

            //根据pod列表更新集群
            this.updateRedisCluster(ns, rediscluster);
        }


    }

    //创建redis cluster集群

    private void createRedisCluster(String ns, Rediscluster rediscluster) throws Exception {
        while (true) {

            Deployment deployment =
                    kubernetesClient
                            .apps()
                            .deployments()
                            .inNamespace(ns)
                            .withName(rediscluster.getMetadata().getName())
                            .get();
            log.info(" {} in {}", rediscluster.getSpec().getPodNum(), deployment.getStatus().getReadyReplicas());
            if (deployment.getStatus().getReadyReplicas() == rediscluster.getSpec().getPodNum()) {

                //查询pod列表
                List<Pod> pods = kubernetesClient
                        .pods()
                        .inNamespace(ns)
                        .withLabel("app", rediscluster.getMetadata().getName())
                        .list().getItems();
                //获取IP列表
                log.info("podNum:{}", pods.size());
                List<String> ipList = new ArrayList<>();
                if (pods != null && !pods.isEmpty()) {
                    for (Pod pod : pods) {
                        log.info("{}:{}", pod.getMetadata().getName(), pod.getStatus().getPodIP());
                        ipList.add(pod.getStatus().getPodIP() + ":6379");
                    }
                }

                //缓存ip列表用于更新
                cacheIPs.put(rediscluster.getMetadata().getName(),ipList);


                //执行创建集群脚本
                //必须加\n否则不执行
                String cmd = "/usr/local/bin/redis-cli --cluster  create -a " + rediscluster.getSpec().getPassword() + " --cluster-yes " + StringUtils.join(ipList, " ") + " --no-auth-warning \n";
                log.info("{},run cmd:{}", pods.get(0).getMetadata().getName(), cmd);

                this.excutePodCommand(kubernetesClient,ns,pods.get(0).getMetadata().getName(),cmd);


                break;
            } else {
                Thread.sleep(1000);
            }
        }
    }

    //更新redis cluster集群

    private void updateRedisCluster(String ns, Rediscluster rediscluster) throws Exception {
        while (true) {

            Deployment deployment =
                    kubernetesClient
                            .apps()
                            .deployments()
                            .inNamespace(ns)
                            .withName(rediscluster.getMetadata().getName())
                            .get();
            log.info(" {} in {}", rediscluster.getSpec().getPodNum(), deployment.getStatus().getReadyReplicas());
            if (deployment.getStatus().getReadyReplicas() == rediscluster.getSpec().getPodNum()) {

                //查询pod列表
                List<Pod> pods = kubernetesClient
                        .pods()
                        .inNamespace(ns)
                        .withLabel("app", rediscluster.getMetadata().getName())
                        .list().getItems();
                //获取IP列表
                log.info("podNum:{}", pods.size());

                List<String> oldIPs =cacheIPs.get(rediscluster.getMetadata().getName());


                String execPodName = "";//选中执行脚本的pod
                String execPodIp ="";
                List<String> ipList = new ArrayList<>();
                List<String> addIpList = new ArrayList<>();
                if (pods != null && !pods.isEmpty()) {
                    for (Pod pod : pods) {
                        log.info("{}:{}", pod.getMetadata().getName(), pod.getStatus().getPodIP());
                        if(oldIPs.contains(pod.getStatus().getPodIP()+ ":6379")){
                            execPodName=pod.getMetadata().getName();
                            execPodIp=pod.getStatus().getPodIP()+ ":6379";
                        } else {
                            addIpList.add(pod.getStatus().getPodIP() + ":6379");
                        }
                        ipList.add(pod.getStatus().getPodIP() + ":6379");
                    }
                }

                //缓存ip列表用于更新
                cacheIPs.put(rediscluster.getMetadata().getName(),ipList);

                //执行增加节点脚本
                //必须加\n否则不执行
                String cmd = "/usr/local/bin/redis-cli --cluster  add-node -a " + rediscluster.getSpec().getPassword() + " --cluster-yes " + StringUtils.join(ipList, " ") + " --no-auth-warning \n";
                log.info("{},run cmd:{}", pods.get(0).getMetadata().getName(), cmd);

                this.excutePodCommand(kubernetesClient,ns,pods.get(0).getMetadata().getName(),cmd);

                Thread.sleep(20000);

                //重新分配slot
                cmd = "/usr/local/bin/redis-cli --cluster rebalance --cluster-threshold 1 --cluster-use-empty-masters "+execPodIp+"  -a " + rediscluster.getSpec().getPassword() + " --cluster-yes " + StringUtils.join(ipList, " ") + " --no-auth-warning \n";
                log.info("{},run cmd:{}", pods.get(0).getMetadata().getName(), cmd);
                this.excutePodCommand(kubernetesClient,ns,pods.get(0).getMetadata().getName(),cmd);

                break;
            } else {
                Thread.sleep(1000);
            }
        }
    }
    //在指定pod执行脚本
    private void excutePodCommand(KubernetesClient client ,String ns,String execPodName,String cmd) {
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        try (
                ExecWatch watch = kubernetesClient.pods().inNamespace(ns).withName(execPodName)
                        .redirectingInput()
                        .redirectingOutput()
                        .redirectingError()
                        .redirectingErrorChannel()
                        .exec();
                InputStreamPumper pump = new InputStreamPumper(watch.getOutput(), new SystemOutCallback())) {

            executorService.submit(pump);
            watch.getInput().write(cmd.getBytes());
            Thread.sleep(5 * 1000L);
        } catch (Exception e) {
            throw KubernetesClientException.launderThrowable(e);
        } finally {
            executorService.shutdownNow();
        }
    }

    //删除创建的k8s deployment
    private void deleteDeployment(Rediscluster rediscluster) throws Exception {
        log.info("Deleting Deployment {}", rediscluster.getMetadata().getName());
        RollableScalableResource<Deployment> deployment =
                kubernetesClient
                        .apps()
                        .deployments()
                        .inNamespace(rediscluster.getMetadata().getNamespace())
                        .withName(rediscluster.getMetadata().getName());
        if (deployment.get() != null) {
            deployment.delete();
        }
    }


    //删除创建的k8s service
    private void deleteService(Rediscluster rediscluster) throws Exception {
        log.info("Deleting Service {}", rediscluster.getMetadata().getName());
        ServiceResource<Service> service =
                kubernetesClient
                        .services()
                        .inNamespace(rediscluster.getMetadata().getNamespace())
                        .withName(rediscluster.getMetadata().getName());
        if (service.get() != null) {
            service.delete();
        }
    }

    //创建或更新k8s service
    private void createOrUpdateService(Rediscluster rediscluster) throws Exception {
        Service service = loadYaml(Service.class, "/redis-cluster-service-template.yml");
        service.getMetadata().setName(rediscluster.getMetadata().getName());
        String ns = rediscluster.getMetadata().getNamespace();
        service.getMetadata().setNamespace(ns);
        service.getSpec().getSelector().put("app", rediscluster.getMetadata().getName());
        service.getSpec().getPorts().get(0).setNodePort(rediscluster.getSpec().getPort());//nodePort外部访问端口
        log.info("Creating or updating Service {} in {}", service.getMetadata().getName(), ns);
        kubernetesClient.services().inNamespace(ns).createOrReplace(service);
    }

    private <T> T loadYaml(Class<T> clazz, String yaml) {
        try (InputStream is = getClass().getResourceAsStream(yaml)) {
            return Serialization.unmarshal(is, clazz);
        } catch (IOException ex) {
            throw new IllegalStateException("Cannot find yaml on classpath: " + yaml);
        }
    }

    private static class SystemOutCallback implements Callback<byte[]> {
        @Override
        public void call(byte[] data) {
            log.info("data:{}", new String(data));
        }
    }
}
